package cn.sc.summer.job.dalayqueue;

import cn.sc.summer.redis.util.RedisHelper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.lang.NonNull;
import org.springframework.stereotype.Component;

import java.util.Iterator;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.DelayQueue;

/**
 * 类名：延时任务队列管理器
 *
 * @author a-xin
 * @date 2023/12/15 10:21
 */
@Slf4j
@Component
public class DelayQueueManager implements CommandLineRunner {

    /**
     * 延迟队列分布式key
     */
    public static final String DELAY_QUEUE = "DELAY_QUEUE:";

    /**
     * 延时任务队列
     */
    private final DelayQueue<DelayTask> delayQueue = new DelayQueue<>();

    /**
     * 加入到延时队列中
     *
     * @param task 任务详情
     */
    public void put(@NonNull DelayTask task) {
        if (task.getNumber() > 0) {
            log.error("==> Add a retry delayed task: {}, current: {}", task.getId(), task.getNumber());
        } else {
            log.info("==> Add a delayed task: {}", task.getId());
        }
        RedisHelper.set(DELAY_QUEUE + task.getId(), System.currentTimeMillis());
        delayQueue.put(task);
    }

    /**
     * 取消延时任务
     *
     * @param task 任务详情
     * @return 是否移除成功
     */
    public boolean remove(DelayTask task) {
        log.error("==> Cancel a delayed task: {}", task.getId());
        RedisHelper.del(DELAY_QUEUE + task.getId());
        return delayQueue.remove(task);
    }

    /**
     * 根据任务ID取消延时任务
     *
     * @param id 任务ID
     * @return 是否取消成功
     */
    public boolean remove(@NonNull String id) {
        Iterator<DelayTask> iterator = delayQueue.iterator();
        if (iterator.hasNext()) {
            DelayTask task = iterator.next();
            if (task.getId().equals(id)) {
                RedisHelper.del(DELAY_QUEUE + task.getId());
                return delayQueue.remove(task);
            }
        }
        return false;
    }

    @Value("${project.delay.retryNum:3}")
    private Integer RETRY_NUM;

    @Value("${project.delay.time:3000}")
    private Integer RETRY_TIME;

    @Value("${project.delay.enabled:true}")
    private Boolean DELAY_ENABLED;

    @Override
    public void run(String... args) {
        log.info("==> Start initialize the delay queue...");
        CompletableFuture.runAsync(this::doTask);
    }

    /**
     * 延时任务执行线程
     */
    private void doTask() {
        while (DELAY_ENABLED) {
            DelayTask task = null;
            try {
                task = delayQueue.take();
                processTask(task);
            } catch (Exception e) {
                remove(task.getId());
                log.error("==> An exception occurs when a delayed task is executed: {}", e.getMessage());
                if (task.getNumber() < RETRY_NUM) {
                    put(new DelayTask(task.getTask(),
                            task.getArg(),
                            task.getDesc(),
                            task.getNumber() + 1,
                            task.getId(),
                            System.currentTimeMillis() + RETRY_TIME));
                }
            }
        }
    }

    /**
     * 内部执行延时任务
     *
     * @param task 任务详情
     */
    private void processTask(DelayTask task) {
        log.info("==> Start the taskId: {}, arg: {}", task.getId(), task.getArg());
        if (RedisHelper.hasKey(DELAY_QUEUE + task.getId())) {
            task.getTask().apply(task.getArg());
            RedisHelper.del(DELAY_QUEUE + task.getId());
        }
    }

}
